package com.wecon.dianping.controller.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @author Weikang  Wang
 * @program: com.wecon.dianping.controller.canal
 * @description: canal的启动器
 * @date 2020/3/26 10:17
 */
public class CanalClient {

    public static String IP = AddressUtils.getHostIp();

    /**
     * canal 端口号
     */
    private static Integer PORT = 11111;

    /**
     * destinations
     */
    private static String DESTINATION = "example";

    private static String USERNAME = "";

    private static String PASSWORD = "";

    public static void main(String[] args) throws InvalidProtocolBufferException {
        // 1.创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(IP, PORT), DESTINATION, USERNAME, PASSWORD);
        // 2.链接到服务
        connector.connect();
        // 3.客户端订阅(不提交，以服务端的配置为准注册)
        connector.subscribe();
        // 4.回滚，断开重取
        connector.rollback();

        while (true) {

            Message message = connector.getWithoutAck(100);
            long messageId = message.getId();
            List<CanalEntry.Entry> entryList = message.getEntries();
            int size = message.getEntries().size();

            if (messageId > 0) {
                System.out.println("messageId:" + messageId + ",size:" + size);
            }
            // 5.遍历取值
            for (CanalEntry.Entry entry : entryList) {
                if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                    continue;
                }
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    switch (rowChange.getEventType()) {
                        case INSERT:
                            System.out.println(entry.getHeader().getTableName() + "执行Insert操作");
                            System.out.println(rowData.getAfterColumnsList());
                            break;
                        case DELETE:
                            System.out.println(entry.getHeader().getTableName() + "执行delete操作");
                            System.out.println(rowData.getAfterColumnsList());
                            break;
                        case UPDATE:
                            System.out.println(entry.getHeader().getTableName() + "执行update操作");
                            System.out.println(rowData.getAfterColumnsList());
                            break;
                        default:
                            break;
                    }
                }

            }
        }
    }

}
